package com.lizhiyu.flink.demo1;


import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.util.Collector;

/***
 * Flink 批处理
 * 简单设置DataSet返回格式数据源，将数据源通过flapMap转换后打印
 * 批处理是分批将数据都读完然后再处理
 * 流处理是一个一个的读，然后读的过程中可以处理
 * 批处理一般处理离线数据，处理速度慢 (高延迟)
 * 流处理一般处理实时数据，效率高
 */
public class A1_Flink_Demo2 {

    public static void main(String[] args) throws Exception {
        //批处理用 ExecutionEnvironment
        ExecutionEnvironment env = ExecutionEnvironment.createLocalEnvironment();
        //并行执行数量
        env.setParallelism(1);
        //设置source  DataSet API已经过时
        DataSet<String> stringDataSet = env.fromElements("haha", "hehe,xixi", "enen,oo");
        stringDataSet.print("处理前");
        DataSet<String> flatMapDataSet = stringDataSet.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String s, Collector<String> collector) throws Exception {
                String[] split = s.split(",");
                for (String word : split) {
                    collector.collect(word);
                }
            }
        });
        flatMapDataSet.print("处理后");
        env.execute("JOB 1");
    }
}
